package com.lzq.mq.broker.slave;

import com.alibaba.fastjson2.JSON;
import com.lzq.mq.common.coder.TcpMsg;
import com.lzq.mq.common.dto.StartSyncReqDTO;
import com.lzq.mq.common.dto.StartSyncRespDTO;
import com.lzq.mq.common.enums.BrokerEventCode;
import com.lzq.mq.common.remote.BrokerNettyRemoteClient;
import lombok.extern.slf4j.Slf4j;

import java.util.UUID;

/**
 * 从节点同步服务
 */
@Slf4j
public class SlaveSyncService {

    private BrokerNettyRemoteClient brokerNettyRemoteClient;

    /**
     * 连接主节点
     * @param address 主节点地址
     * @return 返回连接结果
     */
    public boolean connectMasterBrokerNode(String address) {
        String addressAddr[] = address.split(":");
        String ip = addressAddr[0];
        Integer port = Integer.valueOf(addressAddr[1]);
        try {
            brokerNettyRemoteClient = new BrokerNettyRemoteClient(ip, port);
            brokerNettyRemoteClient.buildConnection();
            return true;
        }catch (Exception e) {
            log.error("连接主节点失败:{}", address, e);
        }
        return false;
    }

    /**
     * 发送开始同步消息
     */
    public void sendStartSyncMsg() {
        StartSyncReqDTO startSyncReqDTO = new StartSyncReqDTO();
        startSyncReqDTO.setMsgId(UUID.randomUUID().toString());
        TcpMsg tcpMsg = new TcpMsg(BrokerEventCode.START_SYNC_MSG.getCode(), JSON.toJSONBytes(startSyncReqDTO));
        TcpMsg startSyncMsgResp = brokerNettyRemoteClient.sendSyncMsg(tcpMsg, startSyncReqDTO.getMsgId());
        StartSyncRespDTO startSyncRespDTO = JSON.parseObject(startSyncMsgResp.getBody(), StartSyncRespDTO.class);
        log.info("开启同步成功:{}", startSyncRespDTO);
    }
}
